﻿using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Reactive.Disposables;

namespace System.Reactive.Linq
{
	public static partial class Observable2
	{
#if !EXPERIMENTAL
#if !WINDOWS_PHONE
		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2>> CombineLatest<T1, T2>(this IObservable<T1> first, IObservable<T2> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2>>>() != null);

			return first.CombineLatest(second, Tuple.Create);
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3>> CombineLatest<T1, T2, T3>(this IObservable<Tuple<T1, T2>> first, IObservable<T3> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3>>>() != null);

			return first.CombineLatest(second, (t, t3) => Tuple.Create(t.Item1, t.Item2, t3));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4>> CombineLatest<T1, T2, T3, T4>(this IObservable<Tuple<T1, T2, T3>> first, IObservable<T4> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4>>>() != null);

			return first.CombineLatest(second, (t, t4) => Tuple.Create(t.Item1, t.Item2, t.Item3, t4));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5>> CombineLatest<T1, T2, T3, T4, T5>(this IObservable<Tuple<T1, T2, T3, T4>> first, IObservable<T5> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5>>>() != null);

			return first.CombineLatest(second, (t, t5) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t5));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6>> CombineLatest<T1, T2, T3, T4, T5, T6>(this IObservable<Tuple<T1, T2, T3, T4, T5>> first, IObservable<T6> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6>>>() != null);

			return first.CombineLatest(second, (t, t6) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t6));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <typeparam name="T7">The type of the tuple's seventh component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>> CombineLatest<T1, T2, T3, T4, T5, T6, T7>(this IObservable<Tuple<T1, T2, T3, T4, T5, T6>> first, IObservable<T7> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>>>() != null);

			return first.CombineLatest(second, (t, t7) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t.Item6, t7));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence by creating a tuple whenever one of the observable sequences produces an element.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <typeparam name="T7">The type of the tuple's seventh component.</typeparam>
		/// <typeparam name="T8">The type of the tuple's eighth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining the latest element from each source using a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7, T8>> CombineLatest<T1, T2, T3, T4, T5, T6, T7, T8>(this IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>> first, IObservable<T8> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7, T8>>>() != null);

			return first.CombineLatest(second, (t, t8) => new Tuple<T1, T2, T3, T4, T5, T6, T7, T8>(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t.Item6, t.Item7, t8));
		}
#endif

		/// <summary>
		/// Merges two or more observable sequences into one observable sequence of lists, each containing the latest values from all observable sequences
		/// whenever one of the sequences produces an element.
		/// </summary>
		/// <typeparam name="TSource">The object that provides notification information.</typeparam>
		/// <param name="sources">An enumerable sequence containing two or more observable sequences to be merged.</param>
		/// <remarks>
		/// <para>
		/// All of the observable sequences must produce at least one element; otherwise, the generated sequence will be empty.  Furthermore, if an observable 
		/// sequence produces more than one element before each of the other observable sequences have produced their first elements, then all of the older 
		/// elements are discarded and will not be included in the generated sequence.  Only the latest elements from each sequence are included.
		/// </para>
		/// <para>
		/// The latest value of an observable sequence is always located in the generated lists at the same index in which that sequence is located in the outer sequence.
		/// For example, the values from the first observable sequence in the outer sequence will always be at index zero (0) in the lists that are generated.
		/// </para>
		/// </remarks>
		/// <returns>An observable sequence containing the result of combining the latest elements of all sources into lists.</returns>
		public static IObservable<IList<TSource>> CombineLatest<TSource>(this IEnumerable<IObservable<TSource>> sources)
		{
			Contract.Requires(sources != null);
			Contract.Ensures(Contract.Result<IObservable<IList<TSource>>>() != null);

			return Observable.Defer(() =>
				{
					var count = 0;
					var completed = false;

					return sources.MarkLast()
						.Select(tuple =>
						{
							count++;
							completed = tuple.Item1;

							return tuple.Item2;
						})
						.ToObservable()
						.CombineLatest()
						.SkipWhile(list => !completed || list.Count < count);
				});
		}
#endif

		/// <summary>
		/// Merges two or more observable sequences into one observable sequence of lists, each containing the latest values from the latest consecutive 
		/// observable sequences whenever one of the sequences produces an element.
		/// </summary>
		/// <typeparam name="TSource">The object that provides notification information.</typeparam>
		/// <param name="sources">An observable sequence containing two or more observable sequences to be merged.</param>
		/// <remarks>
		/// <para>
		/// At least two consecutive observable sequences from the beginning of the outer observable sequence must produce at least one element; otherwise, 
		/// the generated sequence will be empty.  Furthermore, if an observable sequence produces more than one element before each of the consecutive observable 
		/// sequences have produced their first elements, then all of the older elements are discarded and will not be included in the generated sequence.
		/// Only the latest elements from each of the consecutive sequences are included.  As new sequences arrive, the size of the generated lists are increased
		/// to accomodate them if they start producing values.
		/// </para>
		/// <para>
		/// The latest value of an observable sequence is always located in the generated lists at the same index in which that sequence is located in the outer sequence.
		/// For example, the values from the first observable sequence in the outer sequence will always be at index zero (0) in the lists that are generated.
		/// Furthermore, once a generated list contains the value for a particular observable sequence, all subsequent lists will also contain the latest value for that 
		/// sequence.  As a result, the number of items in the generated lists may stay the same or grow when new observable sequences arrive, but the size of the lists 
		/// will never shrink.
		/// </para>
		/// </remarks>
		/// <returns>An observable sequence containing the result of combining the latest elements of all sources into lists.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope",
			Justification = "The disposable is composited by the subscription.")]
		public static IObservable<IList<TSource>> CombineLatest<TSource>(this IObservable<IObservable<TSource>> sources)
		{
			Contract.Requires(sources != null);
			Contract.Ensures(Contract.Result<IObservable<IList<TSource>>>() != null);

			return Observable.Create<IList<TSource>>(
				observer =>
				{
					var gate = new object();

					var latest = new List<TSource>();
					var hasValueFlags = new List<bool>();

					var sourceCount = 0;
					var consecutiveActiveSourcesCount = 0;
					var outerCompleted = false;

					var outerSubscription = new SingleAssignmentDisposable();
					var disposables = new CompositeDisposable(outerSubscription);

					outerSubscription.Disposable = sources.Subscribe(
						source =>
						{
							int index;

							lock (gate)
							{
								sourceCount++;

								index = latest.Count;

								latest.Add(default(TSource));
								hasValueFlags.Add(false);
							}

							var subscription = new SingleAssignmentDisposable();

							disposables.Add(subscription);

							subscription.Disposable = source.Subscribe(
								value =>
								{
									lock (gate)
									{
										latest[index] = value;

										if (consecutiveActiveSourcesCount < hasValueFlags.Count)
										{
											hasValueFlags[index] = true;

											while (consecutiveActiveSourcesCount < hasValueFlags.Count && hasValueFlags[consecutiveActiveSourcesCount])
											{
												consecutiveActiveSourcesCount++;
											}
										}

										if (consecutiveActiveSourcesCount >= 2)
										{
											observer.OnNext(latest.Take(consecutiveActiveSourcesCount).ToList().AsReadOnly());
										}
									}
								},
								observer.OnError,
								() =>
								{
									bool completeNow;

									lock (gate)
									{
										disposables.Remove(subscription);

										sourceCount--;

										completeNow = outerCompleted && sourceCount == 0;
									}

									if (completeNow)
									{
										observer.OnCompleted();
									}
								});
						},
						observer.OnError,
						() =>
						{
							bool completeNow;

							lock (gate)
							{
								outerCompleted = true;

								completeNow = sourceCount == 0;
							}

							if (completeNow)
							{
								observer.OnCompleted();
							}
						});

					return disposables;
				});
		}
	}
}